package com.hudson.admin.client;

import com.hudson.common.domain.InnerMessage;
import com.hudson.common.domain.meta.TopicQueueMap;
import com.hudson.common.enums.MessageTypeEnum;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class BrokerClientManege {
    private static final Logger log = LoggerFactory.getLogger(BrokerClientManege.class);

    private static final ConcurrentHashMap<String, Channel> brokerClientMap = new ConcurrentHashMap<>();

    public static final ConcurrentHashMap<String, CompletableFuture<Map<String, Integer>>> topicQueueMap = new ConcurrentHashMap<>();

    public static void put(String brokerName, Channel channel) {
        brokerClientMap.put(brokerName, channel);
    }

    public static void connect(Map<String, String> brokerAddressMap) {
        List<String> removeList = new ArrayList<>();
        brokerClientMap.forEach((k,v) ->{
            if(!brokerAddressMap.containsKey(k)){
                removeList.add(k);
            }
        });
        for (String brokerName : removeList) {
            close(brokerName);
        }

        for (Map.Entry<String, String> entry : brokerAddressMap.entrySet()) {
            brokerClientMap.computeIfAbsent(entry.getKey(), k -> BrokerClient.connect(entry.getValue()));
        }
    }

    public static Channel getChannel(String brokerName) {
        Channel channel = brokerClientMap.get(brokerName);
        if (channel == null) {
            log.error("此 broker 的 channel 为空");
            return null;
        }
        if (channel.isActive()) {
            return channel;
        } else {
            log.error("此 broker 的 channel 未存活");
            close(brokerName);
            return null;
        }
    }

    public static void close(String brokerName){
        Channel removed = brokerClientMap.remove(brokerName);
        removed.close().addListener(f -> {
            if (f.isSuccess()) {
                log.debug("channel 关闭成功");
            } else {
                log.warn("channel 关闭失败");
            }
        });
    }

    public Map<String,Integer> getTopicMap(String brokerName) {
        Channel channel = brokerClientMap.get(brokerName);
        channel.writeAndFlush(new InnerMessage<>(MessageTypeEnum.QUEUE_MAP.getType(), new TopicQueueMap()));
        CompletableFuture<Map<String,Integer>> future = new CompletableFuture<>();
        topicQueueMap.put(brokerName,future);

        try {
            return future.get();
        } catch (Exception e) {
            log.debug("获取topic队列失败");
            throw new RuntimeException(e);
        }
    }
}
